use std::{fmt::{self}, net::SocketAddr, sync::Arc};

use dashmap::DashMap;
use tokio::{net::{TcpListener, TcpStream}, sync::mpsc};

use anyhow::Result;
use tokio_util::codec::{Framed, LinesCodec};
use tracing::{info, level_filters::LevelFilter, warn};
use tracing_subscriber::{fmt::{Layer}, layer::SubscriberExt, util::SubscriberInitExt, Layer as _};
use futures::{stream::SplitStream, SinkExt, StreamExt};


const MAX_MESSAGES : usize = 128;

// 连接的 叫cliet  ， peers 是

#[derive(Debug, Default)]
pub struct State {
    peers: DashMap<SocketAddr, mpsc::Sender<Arc<Message>>>,
}

// broadcast:  应该有sender



// 定义几种 message
enum Message{

    //Accept(String),
    UserJoind(String),
    UserLeft(String),
    Chat{ 
        sender: String, // sender
        content: String}
}


#[derive(Debug)]
struct Peer {
    username: String,
    stream: SplitStream<Framed<TcpStream, LinesCodec>>,
}


#[tokio::main]
async fn main() -> Result<()>{

    // let layer = Layer::new().with_filter(LevelFilter::INFO);
    // tracing_subscriber::registry().with(layer).init();
    console_subscriber::init();
    let addr = "0.0.0.0:8080";
    let listener = TcpListener::bind(addr).await?; 

    info!("starting chat server on {} ",addr);
    let state = Arc::new(State::default());  // 一份数据， 多个ownership

    loop{ // 在一个loop 里， 不停的启线程吗？
        let (stream, addr) = listener.accept().await?;
        info!("accepted connection from {} ", addr);

        let state_cloned = state.clone();
        tokio::spawn(async move {
           if let Err(e) =  handle_client(state_cloned, addr,stream).await {
               warn!("faile to handle client ,  error => {}" , e);
           }
        });
    }
   // Ok(())
}


async fn handle_client(state: Arc<State>, addr: SocketAddr, stream: TcpStream)  -> Result<()>{
    // frame 化  数据， 知道如何切为一帧一帧的。 这样应用程序 好处理。 否则都是字节流， 不好处理
    let mut stream  =  Framed::new( stream, LinesCodec::new());

    // 按帧发送， 那  codec 就会我我们数据 结尾加 \n
    stream.send("enter your username :").await?;

        
        
        let username =  match stream.next().await {
            Some(Ok(username)) => username,
            Some(Err(e))  => return Err(e.into()),
            None => return Ok(()), // 这里应该走一循环， 一直有输入名字为止
        };        
   

    let mut peer = state.add(addr, stream, username).await;

    let message = Arc::new(Message::user_joined(&peer.username.clone()));

    state.broadcast(addr, message).await;

    while let Some(line) = peer.stream.next().await {
        let line = match line {
            Ok(line) => line,
            Err(e) => {
                warn!( "Failed to read line from {} : {}", addr, e);
                break;
            }
        };

        let message = Arc::new(Message::chat( peer.username.clone(), line ));

        state .broadcast(addr, message).await;
    }

    // when   while loop exit , peer has left the chat  or line reading failed 
    state.peers.remove(&addr);
    // or remove 
    let message = Arc::new(Message::user_left(&peer.username));

    state.broadcast(addr, message).await;
    Ok(())

}




/**
 * addr: 自身的addr， 如果 是自己发的消息， 就不要广播了。
 */
impl State {
     async fn broadcast(&self,addr: SocketAddr, message: Arc<Message>) {
        // 遍历 所有的地址  都发一次， 感觉 add 里可以不用单独发， 这里发就好， message，   不行，要从add 里拿， 拿到消息 不发， 也是可以的吧
        for peer in self.peers.iter() {
            if peer.key() == &addr  {
                continue;
            }
            if let Err(e)  = peer.value().send(message.clone()).await {
                warn!("Failed to broadcase message to {}: {}" , peer.key(), e);
                // remove peer from state, 发送不成功的原因， 一般都是  对应的终端 下线了
                self.peers.remove(peer.key());
            }


        }
     }


     async fn add(&self, addr: SocketAddr, stream : Framed<TcpStream, LinesCodec>, username: String) -> Peer {
        let (tx,mut rx) = mpsc::channel(MAX_MESSAGES);
        self.peers.insert(addr, tx);  // peers 是个 dashboard, 连接上的信息数据

        // ask user for username

       // let username = 

        let (mut stream_sender,stream_receiver) = stream.split();

        // receive messages from others and send them to the client
        tokio::spawn(async move {
            while let Some(message) = rx.recv().await {

                //if let Err(e) 
                if let Err(e) = stream_sender.send(message.to_string()).await {
                    warn!("Failed to send message to {}: {}" , addr,e)   ;
                    break;
                }
                // 这个是群发， 发给指定的addr.  好像不是这意思 ， 一个clinet 一个peer， peer 自己收消息， 发到自己的client 上。  c
                
                
            }

        });


        // return a peer
        Peer {
            username,
            stream: stream_receiver,
        }

     }
}


impl Message {
    fn user_joined(username: &str) -> Self{
        let content = format!("{} has joined the chat ", username);
        Self::UserJoind(content)
    }

    fn user_left(username: &str) -> Self{
        let content = format!("{} has left the chat ", username);
        Self::UserLeft(content)
    }

    fn chat(send: impl Into<String>, content: impl Into<String>) -> Self {

        Self::Chat {sender: send.into(), content:content.into() }
    }
}


impl fmt::Display for Message{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match  self {
            Message::UserJoind(content) =>  write!(f,"[join event: {}]", content),
            Message::UserLeft(content) => write!(f,"[let event: {}]", content),
            Message::Chat { sender, content } => write!(f,"{}: {}", sender, content),
        }
       
    }
}

